package com.mng.stateTest;

import com.mng.dto.SensorReading;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.state.FunctionInitializationContext;
import org.apache.flink.runtime.state.FunctionSnapshotContext;
import org.apache.flink.streaming.api.checkpoint.CheckpointedFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

public class opstate {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        DataStreamSource<String> stringDS = env.socketTextStream("localhost", 7777);
        SingleOutputStreamOperator<SensorReading> sensorStream = stringDS.map(line -> {
            String[] strings = line.split(",");
            return new SensorReading(strings[0], Long.parseLong(strings[1]), Double.parseDouble(strings[2]));

        });
        sensorStream.keyBy(sensorReading -> sensorReading.getId())
                .flatMap(new MyMapfunction(10D))
                .print();
        env.execute();
    }
}

class MyMapfunction extends RichFlatMapFunction<SensorReading, SensorReading> {
    private Double thresold;

    ValueState<Double> lasttemp;

    public MyMapfunction(Double thresold) {
        this.thresold = thresold;
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
        lasttemp = getRuntimeContext().getState(new ValueStateDescriptor<Double>("l_t",Double.class));

    }

    @Override
    public void close() throws Exception {
        super.close();
        lasttemp.clear();
    }

    @Override
    public void flatMap(SensorReading sensorReading, Collector<SensorReading> collector) throws Exception {
        Double lastTempV = lasttemp.value();
        Double nowTemp = sensorReading.getTemperaturre();
        lasttemp.update(nowTemp);
        if(lastTempV != null){
            if(Math.abs(lastTempV-nowTemp)>=this.thresold){
                collector.collect(sensorReading);
            }
        }
    }
}